package com.test.study.ToolsUtils.strom.workdsum;

import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Tuple;

import com.esotericsoftware.minlog.Log;

public class WorkBolt extends BaseRichBolt {
	/**
	 * 
	 */
	private static final long serialVersionUID = 1194717328900087126L;
	@SuppressWarnings({ "rawtypes", "unused" })
	private Map stormConf;
	@SuppressWarnings("unused")
	private TopologyContext context;
	@SuppressWarnings("unused")
	private OutputCollector collector;

	@SuppressWarnings("rawtypes")
	@Override
	public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
		this.stormConf = stormConf;
		this.context = context;
		this.collector = collector;
	}

	HashMap<String, Integer> map = new HashMap<String, Integer>();

	@Override
	public void execute(Tuple input) {
		try {
			String stringByField = input.getStringByField("word");
			if (!stringByField.isEmpty()) {
				Integer value = map.get(stringByField);
				if(null==value) {
					value=0;
				}
				value++;
				map.put(stringByField, value);
			}
			System.out.println("=====================");
			for (Entry<String, Integer> entry : map.entrySet()) {
				System.out.println(entry.getKey()+"\t"+entry.getValue());
				
			}
			this.collector.ack(input);
		} catch (Exception e) {
			Log.error(e.getMessage());
			this.collector.fail(input);
		}
		
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {

	}

}
